package com.ehotting.mqbd.core.helper;


import com.alibaba.fastjson.JSON;
import com.ehotting.mqbd.core.config.MqbdCorePropertyConfig;
import com.ehotting.mqbd.core.manager.MqQueueManager;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.admin.ConsumeStats;
import org.apache.rocketmq.common.admin.OffsetWrapper;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;

@Slf4j
public class RocketMQAdminHelper {

    private static final Logger logger = LoggerFactory.getLogger(MqQueueManager.class);

    private DefaultMQAdminExt defaultMQAdminExt = null;

    private MqbdCorePropertyConfig mqbdCorePropertyConfig;

    public static long EXCEPTION_INDEX = -10000;

    public RocketMQAdminHelper(MqbdCorePropertyConfig mqbdCorePropertyConfig){
        this.mqbdCorePropertyConfig = mqbdCorePropertyConfig;
    }

    public void start(){
        defaultMQAdminExt = new DefaultMQAdminExt();
        defaultMQAdminExt.setNamesrvAddr(mqbdCorePropertyConfig.getProducer().getNameSrvAddr());

        defaultMQAdminExt.setMqClientApiTimeout(mqbdCorePropertyConfig.getConsumer().getMqClientApiTimeout());
        try {
            defaultMQAdminExt.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }

    public long getQueueIndex(String topic,String consumerGroup, String msgId){
        logger.info("Start getQueueIndexByAgentId");
        logger.info("topic:{},consumerGroup:{},msgId:{}",topic,consumerGroup,msgId);
        MessageExt messageExt = null;
        long index = 0;
        try {
            messageExt = defaultMQAdminExt.viewMessage(topic,msgId);
            logger.info("messageExt:{}",JSON.toJSONString(messageExt));
            index = getQueueIndex(topic,consumerGroup,messageExt.getQueueId(),messageExt.getQueueOffset());
        } catch (Throwable e) {
            index = EXCEPTION_INDEX;
            handleException( e, topic, consumerGroup);
        }
        logger.info("index:{}",index);
        logger.info("Finish getQueueIndexByAgentId");
        return index;
    }

    public long getQueueIndex(String topic,String consumerGroup,int queueId,long queueOffset) {
        logger.info("Start getQueueIndexByAgentId");
        logger.info("topic:{},consumerGroup:{},queueId:{},queueOffset:{}",topic,consumerGroup,queueId,queueOffset);
        long queueIndex = 0L;

        try {
            ConsumeStats consumeStats = this.defaultMQAdminExt.examineConsumeStats(consumerGroup, topic);
            List<MessageQueue> mqList = new LinkedList(consumeStats.getOffsetTable().keySet());
            logger.info("mqList.size():{}",mqList.size());
            Iterator iterator = mqList.iterator();
            while(iterator.hasNext()) {
                MessageQueue mq = (MessageQueue)iterator.next();
                logger.info("mq.getTopic():{}", JSON.toJSONString(mq));
                if (topic.equals(mq.getTopic()) && mq.getQueueId()== queueId) {
                    OffsetWrapper offsetWrapper = (OffsetWrapper)consumeStats.getOffsetTable().get(mq);
                    logger.info("offsetWrapper:{}", JSON.toJSONString(offsetWrapper));
                    queueIndex = queueOffset-offsetWrapper.getConsumerOffset();
                }
            }
        } catch (Throwable e) {
            handleException( e, topic, consumerGroup);
            queueIndex = EXCEPTION_INDEX;
        }
        logger.info("queueIndex:{}",queueIndex);
        logger.info("Finish getQueueIndexByAgentId");
        return queueIndex;
    }



    public long getAllQueueSize(String topic,String consumerGroup){
        long allQueueSize = 0L;
        try{
            ConsumeStats consumeStats = defaultMQAdminExt.examineConsumeStats(consumerGroup,topic);
            List<MessageQueue> mqList = new LinkedList(consumeStats.getOffsetTable().keySet());
            for (MessageQueue mq : mqList) {
                //只计算group下此生产端发送对应的Topic
                if (topic.equals(mq.getTopic())) {
                    OffsetWrapper offsetWrapper = (OffsetWrapper) consumeStats.getOffsetTable().get(mq);
                    long diff = offsetWrapper.getBrokerOffset() - offsetWrapper.getConsumerOffset();
                    allQueueSize += diff;
                }
            }
            log.info("allQueueSize:{}" , allQueueSize);
        }catch(Throwable e){
            handleException( e, topic, consumerGroup);
            allQueueSize = 0L;
        }
        return allQueueSize;
    }

    public void stop(){
        if(defaultMQAdminExt!=null){
            defaultMQAdminExt.shutdown();
        }
    }

    private void handleException(Throwable e,String topic,String consumerGroup){
        log.error("**************************START HANDLE ERROR**************************");
        log.error("topic:{},consumerGroup:{}",topic,consumerGroup);
        if(e instanceof  MQClientException ){
            MQClientException ee = (MQClientException)e;
            log.error("CODE:{},DESC:{}",
                    ee.getResponseCode(),
                    ee.getErrorMessage());
            log.error("handleException",e);
        } else if(e instanceof MQBrokerException){
            MQBrokerException ee = (MQBrokerException)e;
            log.error("CODE:{},DESC:{}",
                    ee.getResponseCode(),
                    ee.getErrorMessage());
            log.error("handleException",e);
        }else{
            log.error("handleException", e);
        }
        log.error("**************************FINISH HANDLE ERROR**************************");
    }


}
